package mr.stocks.question1;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import mr.stocks.avro.model.StocksKeyAvro;
import mr.stocks.avro.model.StocksReportAvro;
import mr.stocks.avro.model.StocksValuesAvro;

import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroMultipleOutputs;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class StocksAvroReducer extends
        Reducer<AvroKey<StocksKeyAvro>, AvroValue<StocksReportAvro>, AvroKey<StocksReportAvro>, NullWritable> {
    private final StocksReportAvro resultReport = new StocksReportAvro();
    private AvroKey<StocksReportAvro> outputKey = new AvroKey<StocksReportAvro>();
    private AvroMultipleOutputs out;
    private StringBuilder sb = new StringBuilder();

    @Override
    public void setup(Context context) {
        out = new  AvroMultipleOutputs(context);
    }
    
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        out.close();
    }
    
    @Override
    protected void reduce(AvroKey<StocksKeyAvro> key, Iterable<AvroValue<StocksReportAvro>> incoming, Context context)
            throws IOException, InterruptedException {
        collect(key.datum(), resultReport, incoming);
        outputKey.datum(resultReport);

        sb.setLength(0);
        sb.append(resultReport.getYearMonthDay().getYear().toString() + "_");
        sb.append(resultReport.getYearMonthDay().getMonth().toString() + "_");
        sb.append(resultReport.getYearMonthDay().getDay().toString());
        out.write(outputKey, NullWritable.get(), sb.toString());
      
    }

    public static void collect(StocksKeyAvro key, StocksReportAvro resultReport, Iterable<AvroValue<StocksReportAvro>> incoming) {
        List<StocksValuesAvro> collected = new ArrayList<StocksValuesAvro>();
        for (AvroValue<StocksReportAvro> avroVal : incoming) {
            collected.addAll(avroVal.datum().getStocks());
        }
        
        resultReport.setYearMonthDay(key);
        resultReport.setStocks(collected);
    }
}
